package com.kai.mqtt.session.service.impl;

import com.kai.mqtt.data.PageResult;
import com.kai.mqtt.data.request.PageSessionRequest;
import com.kai.mqtt.data.response.SessionResponse;
import com.kai.mqtt.session.Session;
import com.kai.mqtt.session.service.ISessionService;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author weijk
 * @since 2024/6/9
 */
@Service
public class SessionServiceImpl implements ISessionService {

    private static final Map<String, Session> MAP = new ConcurrentHashMap<>();

    /**
     * 保存会话
     */
    @Override
    public void put(Session session) {
        MAP.put(session.getClientId(), session);
    }

    /**
     * 删除会话
     *
     * @param clientId 客户端 id
     */
    @Override
    public void remove(String clientId) {
        MAP.remove(clientId);
    }

    /**
     * 获取会话信息
     *
     * @param clientId 客户端 id
     */
    @Override
    public Session get(String clientId) {
        return MAP.get(clientId);
    }

    @Override
    public Integer getConnect() {
        return MAP.size();
    }

    @Override
    public Integer getOnlineConnect() {
        return (int) MAP.values().stream().filter(session -> session.getChannel().isActive()).count();
    }

    /**
     * 分页查询会话列表
     *
     * @param request 参数
     * @return 会话列表
     */
    @Override
    public PageResult<SessionResponse> pageSession(PageSessionRequest request) {
        List<Session> sessions = MAP.values().stream()
                .filter(session -> !StringUtils.hasText(request.getClientId()) || session.getClientId().contains(request.getClientId()))
                .sorted(Comparator.comparing(Session::getConnectTime).reversed())
                .toList();

        List<SessionResponse> responses = sessions.stream()
                .skip((long) request.getPageSize() * (request.getPageIndex() - 1))
                .limit(request.getPageSize())
                .map(session -> {
                    SessionResponse sessionResponse = new SessionResponse();
                    BeanUtils.copyProperties(session, sessionResponse);

                    return sessionResponse;
                }).toList();

        return PageResult.success(responses, request.getPageIndex(), request.getPageSize(), sessions.size());
    }

    /**
     * 删除会话
     *
     * @param clientId 客户端 id
     * @return true-成功，false-失败
     */
    @Override
    public Boolean deleteSession(String clientId) {
        Session session = MAP.get(clientId);
        if (Objects.isNull(session)) {
            return Boolean.TRUE;
        }

        session.setIsCleanSession(true);
        session.getChannel().close();

        return Boolean.TRUE;
    }
}
